package com.king.first.app.pv;

import com.king.first.bean.UserBehavior;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class Project_PV_2 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env
                .readTextFile("input/UserBehavior.csv")
                .map(line -> { // 对数据切割, 然后封装到POJO中
                    String[] split = line.split(",");
                    return new UserBehavior(
                            Long.valueOf(split[0]),
                            Long.valueOf(split[1]),
                            Integer.valueOf(split[2]),
                            split[3],
                            Long.valueOf(split[4]));
                })
                .filter(behavior -> "pv".equals(behavior.getBehavior())) //过滤出pv行为
                .map(behavior -> Tuple2.of("pv", 1L)).returns(Types.TUPLE(Types.STRING, Types.LONG)) // 使用Tuple类型, 方便后面求和
                .keyBy(value -> value.f0)  // keyBy: 按照key分组
                .sum(1) // 求和
                .print();

        env.execute();

    }

    public static void main2(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> streamSource = env.readTextFile("input/UserBehavior.csv");

        //894923,3076029,1879194,pv,1511658000

        SingleOutputStreamOperator<Tuple2<UserBehavior, Integer>> flatMap = streamSource.flatMap(new FlatMapFunction<String, Tuple2<UserBehavior, Integer>>() {
            private Integer count = 0;
            @Override
            public void flatMap(String s, Collector<Tuple2<UserBehavior, Integer>> collector) throws Exception {
                String[] split = s.split(",");
                if ("pv".equals(split[3])) {
                    count++ ;
                    collector.collect(Tuple2.of(new UserBehavior(Long.parseLong(split[0]), 0L, 0, split[3], 0L), 1));
                }
            }
        });

        KeyedStream<Tuple2<UserBehavior, Integer>, Integer> tuple2IntegerKeyedStream = flatMap.keyBy(new KeySelector<Tuple2<UserBehavior, Integer>, Integer>() {
            @Override
            public Integer getKey(Tuple2<UserBehavior, Integer> userBehaviorIntegerTuple2) throws Exception {
                return 1;
            }
        });

        SingleOutputStreamOperator<Tuple2<UserBehavior, Integer>> sum = tuple2IntegerKeyedStream.sum(1);

        sum.print();

        env.execute();
    }
}
